/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ocr.send;

import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import cn.ac.iie.di.ocr.handler.Message;
import cn.ac.iie.di.ocr.util.GlobalParameters;
import java.util.LinkedList;
import java.util.NoSuchElementException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.log4j.Logger;

/**
 *
 * @author Administrator
 */
public class PicSendHandler implements Runnable {

    static private final Logger logger = Logger.getLogger(PicSendHandler.class.getName());

    ThreadPoolExecutor threadpool = null;

    public PicSendHandler() {
        //创建一个线程池
        threadpool = new ThreadPoolExecutor(GlobalParameters.coreThreadPoolSize, GlobalParameters.maxThreadPoolSize, 3, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(GlobalParameters.blockingQueueSize),
                new ThreadFactory() {
            AtomicInteger idx = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "PicSendThreadPoolTask_" + idx.getAndIncrement());
            }
        }, new RejectedExecutionHandler() {

            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                while (true) {
                    try {
                        executor.getQueue().put(r);
                        break;
                    } catch (InterruptedException ex) {
                        logger.warn("", ex);
                    }
                }
            }
        });
    }

    @Override
    public void run() {
        RESendSession session = null;
        SessionPool pool;
        while (true) {
            if (GlobalParameters.colNames.length != 0) {
                try {
                    //初始化RESendSessionBuild
                    REConnection conn = new REConnection(GlobalParameters.nameserver);
                    RESendSessionBuilder builder = (RESendSessionBuilder) conn.getSendSessionBuilder(GlobalParameters.pic_to_topic);

                    for (int i = 0; i < GlobalParameters.colNames.length; i++) {
                        if (!GlobalParameters.colNames[i].equals(GlobalParameters.addColName)) {
                            builder.addColumn(GlobalParameters.colNames[i], GlobalParameters.types[i], true);
                        }
                    }

                    pool = new SessionPool(builder);
                    session = (RESendSession) builder.build();
                    session.start();

                } catch (Exception e) {
                    logger.error("session.build error,cause " + e.getMessage(), e);
                    return;
                }
                logger.info("init RESendSession finish.");
                break;
            } else {
                logger.info("do not init RESendSession. Cause picSendQueue has no data!");
                try {
                    Thread.sleep(500);
                } catch (InterruptedException ex) {
                    logger.error(ex, ex);
                }
            }
        }

        while (true) {
            try {
                Message piclist = (Message) GlobalParameters.picSendQueue.poll(GlobalParameters.intervaltime, TimeUnit.MILLISECONDS);
                if (piclist == null) {
                    logger.info("picSendQueue doesn't hava available data, and sleep 500 ms.");
                    Thread.sleep(500);
                } else {
                    threadpool.execute(new PicSendThreadPoolTask(piclist, pool));
                }
            } catch (Exception ex) {
                logger.error("get JSONData from ArrayBlockingQueue failed,cause :" + ex.getMessage(), ex);
            }
        }

    }

    static public class SessionPool {

        RESendSessionBuilder builder;
        LinkedList<RESendSession> pool;

        public SessionPool(RESendSessionBuilder builder) {
            this.builder = builder;
            pool = new LinkedList<>();
        }

        public synchronized RESendSession borrowSession() throws RESessionException, REConnectionException {
            try {
                return pool.removeFirst();
            } catch (NoSuchElementException ex) {
                RESendSession rtn = (RESendSession) builder.build();
                rtn.start();
                return rtn;
            }
        }

        public synchronized void returnSession(RESendSession session) {
            pool.add(session);
        }
    }

}
